package com.tang.sharding.schedule.task.manager;

import com.tang.sharding.schedule.task.AbstractRunnableTask;
import com.tang.sharding.schedule.task.jdbc.AbstractJdbcRunnableTask;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.config.*;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * @author Tang
 * @classname TaskManager
 * @description [  ]
 * @date 2021/8/2 13:47
 */
@Slf4j
public abstract class AbstractTaskManager implements TaskManager {

    /**
     * 注册器
     */
    @Getter
    protected ScheduledTaskRegistrar scheduledTaskRegistrar;

    /**
     * 任务维护表
     */
    @Getter
    protected Map<String, ScheduledTask> scheduledTaskMap = new HashMap<>(8);

    /**
     * 任务Queue
     */
    @Setter
    @Getter
    protected static BlockingQueue<Object> taskBlockingQueue = new ArrayBlockingQueue<>(16);

    /**
     * 原生任务
     */
    @Getter
    protected Set<ScheduledTask> scheduledTasks;

    public AbstractTaskManager(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        this.scheduledTaskRegistrar = scheduledTaskRegistrar;
        this.scheduledTasks = new HashSet<>(scheduledTaskRegistrar.getScheduledTasks());
        init();
    }

    @Override
    public void addTask(AbstractRunnableTask runnableTask, Task task) {
        if (task instanceof CronTask) {
            CronTask cronTask = (CronTask) task;
            scheduledTaskRegistrar.addCronTask(cronTask);
            ScheduledTask scheduledTask = scheduledTaskRegistrar.scheduleCronTask(cronTask);
            scheduledTasks.add(scheduledTask);
            scheduledTaskMap.put(runnableTask.getTaskId(), scheduledTask);
        }
        if (task instanceof IntervalTask) {
            // do some thing
        }
    }

    /**
     * 取消任务
     *
     * @param taskId
     */
    @Override
    public void cancelTask(String taskId) {
        ScheduledTask scheduledTask = scheduledTaskMap.remove(taskId);
        if (scheduledTask != null) {
            scheduledTasks.remove(scheduledTask);
            scheduledTask.cancel();
        }
    }

    /**
     * 任务ids,列表
     *
     * @return
     */
    public Set<String> taskIds() {
        return this.scheduledTaskMap.keySet();
    }

    /**
     * task 消费者
     */
    private void init() {
        Collection<Object> taskList = new ArrayList<>(5);
        Thread thread = new Thread(() -> {
            log.info(Thread.currentThread().getName() + ",任务消费者线程监听启动...");
            while (true) {
                taskBlockingQueue.drainTo(taskList);
                taskList.stream().filter(task -> {
                    return task instanceof Task;
                }).forEach(task -> {
                    Task task1 = (Task) task;
                    AbstractJdbcRunnableTask jdbcRunnableTask = (AbstractJdbcRunnableTask) task1.getRunnable();
                    log.info("开始启动[task Id为:{},表名为:{}]的任务...",jdbcRunnableTask.getTaskId(),jdbcRunnableTask.getTableRule().getLogicTable());
                    addTask(jdbcRunnableTask, task1);
                });
                taskList.clear();
            }
        }, "task-consumer-thread");
        thread.setDaemon(true);
        thread.start();
    }
}
